package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.*;
import org.apache.flink.types.Row;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import static org.apache.flink.table.api.Expressions.$;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/12 9:30
 */
public class Flink04_TableAPI_Connector_KafkaSource {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO - Connector外部系统，读 Kafka
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv
                .connect(new Kafka()
                        .topic("flink0923")
                        .version("universal")
                        .property(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092,hadoop104:9092")
                        .property(ConsumerConfig.GROUP_ID_CONFIG, "abcd")
                )
//                .withFormat(new Csv())
                .withFormat(new Json())
                .withSchema(new Schema()
                                .field("id", DataTypes.STRING())
                                .field("ts", DataTypes.BIGINT())
                                .field("vc", DataTypes.INT())
                )
                .createTemporaryTable("kafkaSource");

        Table kafkaSource = tableEnv.from("kafkaSource");

        tableEnv.toAppendStream(kafkaSource, Row.class).print();


        env.execute();
    }
}

/**
 * 通过 connector 将外部系统抽象成 动态表
 * 1、表环境.connect(外部系统描述器，指定相关参数)
 * 2、指定数据的存储格式
 * 3、指定 表结构
 * 4、指定 临时表的名字
 */
